package cz.cvut.fel.persister;

import java.io.IOException;
import java.io.StringReader;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.TokenStream;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.analysis.tokenattributes.TermAttribute;
import org.apache.lucene.util.Version;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.basho.riak.pbc.RiakObject;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.Mongo;
import com.mongodb.WriteConcern;
import com.mongodb.gridfs.GridFS;
import com.mongodb.gridfs.GridFSInputFile;

import cz.cvut.fel.email.Email;
import cz.cvut.fel.loader.Persister;

public class MongoDBPersister extends Persister {
	final static Logger log = LoggerFactory.getLogger(MongoDBPersister.class);
	private Mongo mongo;
	private String database = "DOMAIN";
	private DB db;
	private DBCollection coll;
	
	public MongoDBPersister(String server, Integer port) {
		super(parseServer(server), (port == null ? parsePort(server) : port));
		
		
	}

	@Override
	public void connect() throws Exception {
		if (port != null && port != 0)
			mongo = new Mongo(server,port);
		else
			mongo = new Mongo(server);

		db = mongo.getDB(database);
		coll = db.getCollection("DOMAIN");
	}

	/**
	 * @return the coll
	 */
	public DBCollection getColl() {
		return coll;
	}

	/**
	 * @param coll the coll to set
	 */
	public void setColl(DBCollection coll) {
		this.coll = coll;
	}

	@Override
	public void disconnect() {
		db = null;
		mongo.close();

	}

	public void store(String key, byte[] data) throws Exception {
		BasicDBObject bson = new BasicDBObject ();
		bson.put("_id", key);
		bson.put("data",data);

		coll.insert(bson,WriteConcern.REPLICAS_SAFE);
	}

	

	/**
	 * @param database the database to set
	 */
	public void setDatabase(String database) {
		this.database = database;
	}

	/**
	 * @return the database
	 */
	public String getDatabase() {
		return database;
	}

	static Analyzer analyzer = new StandardAnalyzer(Version.LUCENE_30);
	private List<String> tokenize(TokenStream stream) throws IOException {
		List<String> tokens = new ArrayList<String>();
		TermAttribute term = (TermAttribute) stream
				.addAttribute(TermAttribute.class);
		while (stream.incrementToken()) {
			// Not sure if we somehow can use termBuffer() to get a char[]
			// so we do no have to create a new String for each term
			tokens.add(term.term());
		}
		return tokens;
	}
	@Override
	public void store(Email email) throws Exception {
		if (coll == null)
			connect();		
		ArrayList<String> keys = new ArrayList<String>();
		//	log.info(email.getHeaders().toString());
			for (int i = 0; i < email.getAttachments().size(); i++) {
				try {
					keys.add(email.getKeys().get(i));
					
//					GridFS gridFS = new GridFS(db);
//					GridFSInputFile in = gridFS.createFile(email.getAttachments().get(i).getBytes());
					//in.setFilename(email.getK//);
//					in.save();
					//RiakObject obj	= new RiakObject("",email.getKeys().get(i), email.getAttachments().get(i).getBytes());				
					//riakClient.store(obj);
					//log.info("STORED "+email.getKeys().get(i));
				} catch (Exception e) {
					log.error("Unable to store email " + email, e);

				}
			}
			
			if (email.getContent() != null) {
				try {
					HashMap<String,String> headers = email.getHeaders();
					String mailKey = URLEncoder.encode(email.getHeaders().get("MESSAGE-ID"),"UTF-8");
					BasicDBObject bson = new BasicDBObject ();

					try {
						bson.put("FROM", headers.get("X-VF-Scanner-Mail-From".toUpperCase()).split(" ")[0]);
					} catch(Exception e) {
						bson.put("FROM", "dummy@dummy.com");
					}
					try {
						bson.put("MAIL-TO", headers.get("X-VF-Scanner-Rcpt-To".toUpperCase()).split(",")[0]);
					} catch(Exception e) {
						bson.put("MAIL-TO", "dummy@dummy.com");
					}
					
					bson.put("DATE_date", headers.get("DATE"));
					bson.put("SUBJECT", headers.get("SUBJECT"));
					bson.put("DATA", email.getContent());
					bson.toString().length();
					
					bson.append("fulltext", tokenize(analyzer.tokenStream("fulltext", new StringReader( email.getContent()))));
				//	System.out.println(bson.toString());
					if (coll == null)
						connect();
					coll.save(bson,WriteConcern.REPLICAS_SAFE);
					
					//RiakObject obj	= new RiakObject("DOMAIN",mailKey,  bson.toString());
					//obj.setContentType("text/x-json");
					
					
					
					
					
					//log.info("STORED PRIMARY KEY="+mailKey+" ATT COUNT="+email.getAttachments().size());
				} catch (Exception e) {
					log.error("Unable to store email " + email, e);

				}
			}
		
	}

	@Override
	public byte[] readBytes() throws Exception {
		// TODO Auto-generated method stub
		return null;
	}

}
